#include <iterator>
#include <unistd.h>
#include "MarkupSTL.h"
#include "global.h"
#include "cmd.h"
#include "Game.h"
#include "RedisClient.h"
#include "Logic.h"
#include "json.h"
#include "globalExtern.h"
#include "Host.h"
#include "2850_gamerecord.pb.h"


#define  MOD_UID (100)

char* CGameServer::s_szexchangetype = "direct";
char* CGameServer::s_szexchange = "gamerecord_flow";

CGameServer::CGameServer():m_mqPort(0), m_mqconn(NULL), m_mqchannel(1)
{
}

CGameServer::~CGameServer()
{
// 	if (m_mqconn1)
// 		releasemq1();
	if (m_mqconn)
		releasemq();

}

bool CGameServer::InitGame()
{
	char szconfig[] = "./config.xml";
	if (!ReadCommonConf(szconfig))
		return false;

	return initmq();
}

bool CGameServer::initmq(void)
{
	m_mqconn = amqp_new_connection();
	if (!m_mqconn)
	{
		log_error("create mq conn err");
		return false;
	}
	amqp_socket_t* socket = amqp_tcp_socket_new(m_mqconn);
	if (!socket)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("create mq TCP socket err");
		return false;
	}
	int status = amqp_socket_open(socket, m_mqIp.c_str(), m_mqPort);
	if (status)
	{
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq socket err");
		return false;
	}
	amqp_rpc_reply_t x = amqp_login(m_mqconn, m_mqVhost.c_str(), 0, 131072, 0, AMQP_SASL_METHOD_PLAIN, m_mqUsername.c_str(), m_mqPwd.c_str());
	if (AMQP_RESPONSE_NORMAL != x.reply_type)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("mq login fail,host:%s,user:%s,pwd:%s", m_mqVhost.c_str(), m_mqUsername.c_str(), m_mqPwd.c_str());
		return false;
	}
	amqp_channel_open_ok_t* t = amqp_channel_open(m_mqconn, m_mqchannel);
	if (!t)
	{
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
		log_error("open mq channel fail");
		return false;
	}
	// 	if (0 == t->channel_id.len || !t->channel_id.bytes)
	// 	{
	// 		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
	// 		amqp_destroy_connection(m_mqconn);
	// 		m_mqconn = NULL;
	// 		log_error("open mq channel fail");
	// 		return false;
	// 	}
	// 	log_debug("mq channle:%s", std::string((char *)t->channel_id.bytes, t->channel_id.len).c_str());

	amqp_rpc_reply_t x1 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x1.reply_type)
	{
		releasemq();
		log_error("mq open channel amqp_get_rpc_reply fail");
		return false;
	}

	amqp_exchange_declare(m_mqconn, m_mqchannel, amqp_cstring_bytes(CGameServer::s_szexchange), amqp_cstring_bytes(s_szexchangetype), false, true, false, false, amqp_empty_table);

	amqp_rpc_reply_t x2 = amqp_get_rpc_reply(m_mqconn);
	if (AMQP_RESPONSE_NORMAL != x2.reply_type)
	{
		releasemq();
		log_error("mq open amqp_exchange_declare fail");
		return false;
	}

	return true;
}


void CGameServer::releasemq(void)
{
	if (m_mqconn)
	{
		amqp_channel_close(m_mqconn, m_mqchannel, AMQP_REPLY_SUCCESS);
		amqp_connection_close(m_mqconn, AMQP_REPLY_SUCCESS);
		amqp_destroy_connection(m_mqconn);
		m_mqconn = NULL;
	}

	return;
}

int CGameServer::OnBYTimeOut(int Timerid)
{
    switch(Timerid)
	{
    }
	return 0;
}


int CGameServer::ProcessPacket(SocketHandler * pHandler, NETInputPacket* inputPkg)
{
	unsigned int cmd = inputPkg->GetCmdType();
	switch (cmd)
	{
	case gamerecord::GAMERECORD_CMD_RECORD_PUSH:
	{
		procGameRecordFlow(pHandler, inputPkg);
		break;
	}
	default:
		break;
	}

	return 0;
}

int CGameServer::procGameRecordFlow(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	procGameRecordMQ(pHandler, inputPkg);

	return 0;
}

int CGameServer::procGameRecordMQ(SocketHandler* pHandler, NETInputPacket* inputPkg)
{
	gamerecord::GameRecordPush msg;
	if (!inputPkg->parsePb(&msg))
	{
		log_error("parse pb fail");
		return -1;
	}
	int count = msg.list_size();
	for (int i = 0; i < count; i++)
	{
		unsigned int uid = msg.list(i).uid();
		if (uid > 0)
		{
			if (!m_mqconn)
			{
				if (initmq())
				{
					log_alert("mq reconnect OK");
				}
				else
				{
					log_error("mq reconnect fail");
				}
			}

			std::string str;
			msg.list(i).SerializeToString(&str);
			log_debug("procGameRecordMQ,uid:%u,data len:%u", uid, str.size());

			if (m_mqconn)
			{
				amqp_basic_properties_t props;
				props._flags = AMQP_BASIC_DELIVERY_MODE_FLAG;
				props.delivery_mode = 2; /* persistent delivery mode */

				unsigned int index = uid % MOD_UID;
				amqp_bytes_t message;
				message.bytes = const_cast<char*>(str.data());
				message.len = str.size();
				int ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexchange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
				log_debug("procGameRecordMQ,uid:%u,data len:%u,ret:%x", uid, str.size(), ret);
				if (AMQP_STATUS_OK != ret)
				{
					log_error("amqp_basic_publish fail,ret:%d,msg len:%u", ret, message.len);
					if (AMQP_STATUS_CONNECTION_CLOSED == ret || AMQP_STATUS_SOCKET_ERROR == ret)
					{
						releasemq();
						if (initmq())
						{
							ret = amqp_basic_publish(m_mqconn, m_mqchannel, amqp_cstring_bytes(s_szexchange), amqp_cstring_bytes(std::to_string(index).c_str()), 0, 0, &props, message);
							log_alert("mq reconnect OK, publish ret:%d", ret);
						}
						else
						{
							log_error("reconnect fail,ret:%d", ret);
						}
					}
				}
			}

		}
	}

	return 0;
}

bool CGameServer::ReadCommonConf(const char* strxml)
{
	CMarkupSTL  markup;
	if (!markup.Load(strxml))
	{
		log_error("Load xml %s failed", strxml);
		return false;
	}

	if (!markup.FindElem("application"))
	{
		log_error("Not Found Elem [application] in xml[%s]", strxml);
		return false;
	}

	if (!markup.IntoElem())
	{
		log_error("IntoElem [application] failed.");
		return false;
	}


	if (!markup.FindElem("mq"))
	{
		log_error("Not Found Elem [mq] in xml[%s]", strxml);
		return false;
	}
	else
	{
		m_mqIp = markup.GetAttrib("ip");
		m_mqPort = atoi(markup.GetAttrib("port").c_str());
		m_mqVhost = markup.GetAttrib("vhost");
		m_mqUsername = markup.GetAttrib("username");
		m_mqPwd = markup.GetAttrib("password");
	}

	if (m_mqIp.empty() || m_mqVhost.empty() || m_mqUsername.empty() || m_mqPwd.empty() || 0 == m_mqPort)
		return false;

	log_debug("ReadCommonConf succ!!");

	return true;
}